package ex.tableapi.window;

import ex.tableapi.ApiFrame;
import org.apache.flink.table.api.Over;
import org.apache.flink.table.api.Table;

import static org.apache.flink.table.api.Expressions.$;
import static org.apache.flink.table.api.Expressions.UNBOUNDED_RANGE;

public class ExOverWindows extends ApiFrame {
    private String outName = "printOutTable";

    public static void main(String[] args) {
        ExOverWindows demo = new ExOverWindows();
        demo.getEnv();
        Table orders = demo.registerMysql("myorder", "orders");
        Table result = demo.exOverWindows(orders);

        demo.registerConsole(demo.createPrintOutDDL(result.getResolvedSchema().toString()), demo.outName);
        result.executeInsert(demo.outName);
    }

    private Table exOverWindows(Table orders) {
//        Table result = orders.window(Session.withGap(lit(10).minutes()).on($("rowtime")).as("w"))
//                .groupBy($("a"), $("w"))
//                .select($("a"), $("b").sum().over($("w"))//, $("c").min().over($("w"))
//                );
// 按a,c分区，c组数据依次累计求和。
        Table result = orders
                .window(Over
                        .partitionBy($("a"), $("c"))
                        .orderBy($("rowtime"))
                        .preceding(UNBOUNDED_RANGE)
                        .as("ow")) // 注意：此处不能使用 Expression 表达式
                .select($("a"),
                        $("c"),
                        $("rowtime"),
                        $("b").sum().over($("ow")));
        return result;
    }



}
